package com.idanchuang.support.etcache.pubsub.pulsar;

import com.idanchuang.support.etcache.pubsub.AbstractRemoteEventSubscriber;
import com.idanchuang.support.etcache.pubsub.SubEventQueue;
import com.idanchuang.support.etcache.remote.codec.RemoteKeyCodec;
import com.idanchuang.support.etcache.util.InetUtil;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageListener;
import org.apache.pulsar.client.api.PulsarClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.management.ManagementFactory;

/**
 * 基于 Pulsar 实现的缓存变更事件监听器
 * @author yjy
 * Created at 2021/12/22 12:44 下午
 */
public class PulsarRemoteEventSubscriber extends AbstractRemoteEventSubscriber implements MessageListener<byte[]> {

    private static final Logger logger = LoggerFactory.getLogger(PulsarRemoteEventSubscriber.class);

    private final PulsarProperties properties;
    private final String topic;
    private final String subscriptionName;
    private PulsarClient client;
    private Consumer<byte[]> consumer;
    private volatile boolean shutdown = false;

    public PulsarRemoteEventSubscriber(RemoteKeyCodec remoteKeyCodec,
                                       SubEventQueue subEventQueue,
                                       String env, String appId, PulsarProperties properties) {
        super(remoteKeyCodec, subEventQueue);
        try {
            this.properties = properties;
            this.topic = env + "-base/etcache/event-" + appId.toLowerCase();
            String localIp = InetUtil.getLocalIp();
            String pid = ManagementFactory.getRuntimeMXBean().getName().split("@")[0];
            this.subscriptionName = appId + ":" + localIp + ":" + pid;
            this.client = PulsarClient.builder().serviceUrl(properties.getServiceUrl()).build();
            this.init();
        } catch (Exception e) {
            throw new RuntimeException("PulsarRemoteEventSubscriber init failed", e);
        }
    }

    private void init() throws Exception {
        this.consumer = client.newConsumer()
                .topic(this.topic)
                .subscriptionName(subscriptionName)
                .messageListener(this)
                .subscribe();
        logger.info("PulsarRemoteEventSubscriber init finish > topic: {}, subscriptionName: {}",
                topic, subscriptionName);
    }

    @Override
    public void destroy() {
        try {
            if (!this.shutdown) {
                this.shutdown = true;
                if (this.client != null && !this.client.isClosed()) {
                    this.client.shutdown();
                }
                logger.info("PulsarRemoteEventSubscriber destroy finish > topic: {}, subscriptionName: {}",
                        topic, subscriptionName);
            }
        } catch (Exception e) {
            logger.warn("pulsar client close failed, cause: {}", e.getMessage());
        }
    }

    /**
     * 重连amqp
     */
    private void reconnect() {
        while (!this.consumer.isConnected() && !shutdown) {
            try {
                Thread.sleep(1000L);
                if (this.client.isClosed()) {
                    this.client = PulsarClient.builder().serviceUrl(properties.getServiceUrl()).build();
                }
                this.init();
            } catch (InterruptedException e) {
                logger.warn("reconnect pulsar cancel, cause thread is interrupted", e);
                break;
            } catch (Exception e) {
                logger.error("reconnect pulsar failed, will retry after 1s..", e);
            }
        }
    }

    @Override
    public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
        try {
            handle(msg.getData());
            consumer.acknowledge(msg);
        } catch (Exception e) {
            consumer.negativeAcknowledge(msg);
        }
    }

    @Override
    public void reachedEndOfTopic(Consumer<byte[]> consumer) {
        reconnect();
    }
}
